package day03.acc

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @Author wsl
 * @Description
 * 血缘：toDebugString  一个RDD是从哪些RDD转换而来的
 * 依赖：dependencies   子RDD依赖父RDD的哪些分区
 *
 * stage划分：
 * （1）Application：初始化一个SparkContext即生成一个Application；
 * （2）Job：一个Action算子就会生成一个Job；
 * （3）Stage：Stage等于宽依赖的个数加1；
 * （4）Task：一个Stage阶段中，最后一个RDD的分区个数就是Task的个数
 *
 */
object Lineage {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("lineage").setMaster("local[*]")
    //Application：初始化一个SparkContext即生成一个Application
    val sc = new SparkContext(conf)

    // 继承了HadoopRDD[0]  自己是MapPartitionsRDD[1]
    val lineRDD: RDD[String] = sc.textFile("sparkcore/input/1.txt")
    println(lineRDD.toDebugString)
    println(lineRDD.dependencies)
    println("----------------------------------------------")


    // HadoopRDD[0]  MapPartitionsRDD[1]   MapPartitionsRDD[2]
    val wordRDD: RDD[String] = lineRDD.flatMap(_.split(" "))
    println(wordRDD.toDebugString)
    println(wordRDD.dependencies)
    println("----------------------------------------------")


    // HadoopRDD[0]  MapPartitionsRDD[1]  MapPartitionsRDD[2] MapPartitionsRDD[3]
    // OneToOneDependency
    val word2oneRDD: RDD[(String, Int)] = wordRDD.map((_, 1))
    println(word2oneRDD.toDebugString)
    println(word2oneRDD.dependencies)
    println("----------------------------------------------")

    //textFile,flatMap,map算子全部是窄依赖,不会增加stage阶段
    // //reduceByKey算子会有宽依赖,stage阶段加1，2个stage

    // HadoopRDD[0]  MapPartitionsRDD[1]  MapPartitionsRDD[2] MapPartitionsRDD[3]  ShuffledRDD[4]
    // ShuffleDependency  spark调优避免使用宽依赖
    val word2sumRDD: RDD[(String, Int)] = word2oneRDD.reduceByKey(_ + _)
    println(word2sumRDD.toDebugString)
    println(word2sumRDD.dependencies)
    //class OneToOneDependency[T](rdd: RDD[T]) extends NarrowDependency
    println("----------------------------------------------")

    //Job：一个Action算子就会生成一个Job，2个Job
    word2sumRDD.collect().foreach(println)
    word2sumRDD.saveAsTextFile("/out4")

    //阻塞线程,方便进入localhost:4040历史服务器查看
    Thread.sleep(Long.MaxValue)

    sc.stop()
  }
}
